"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 解码子事务逻辑复制
Create At   : 2022/03/01
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.创建用户，修改pg_hba文件
    4.修改逻辑复制相关参数
    5.创建表，创建逻辑复制槽
    6.开启逻辑复制槽
    7.DML操作
    8.比对数据
    9.停止逻辑复制槽
    10.删除逻辑复制槽
    11.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.创建成功，修改pg_hba文件成功
    4.修改逻辑复制相关参数成功
    5.创建表，创建逻辑复制槽成功
    6.开启逻辑复制槽成功
    7.DML操作执行成功
    8.比对数据成功
    9.停止逻辑复制槽成功
    10.删除逻辑复制槽成功
    11.资源清理成功，恢复参数成功
History     :
"""
import os
import time
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.standby_node = Node('Standby1DbUser')
        self.standby_root = Node('Standby1Root')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0001/')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.user_name = 'u_standby_logic_replication_0001'
        self.table_name = 't_standby_logic_replication_0001'
        self.function_name = 'insert_to_table_0001'
        self.procedure_name = 'pro_standby_logic_replication_0001'
        self.slot_name_1 = 'slot_standby_logic_replication_0001_1'
        self.slot_name_2 = 'slot_standby_logic_replication_0001_2'
        self.log_file_1 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0001_1')
        self.log_file_2 = os.path.join(self.dir_replication,
                                       'log_standby_logic_replication_0001_2')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_primy_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_primy_msg)
        self.assertEqual("OK", cp_primy_msg, "执行失败" + text)
        cp_standby_msg = self.standby_node.sh(cp_cmd).result()
        self.logger.info(cp_standby_msg)
        self.assertEqual("OK", cp_standby_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:创建用户,修改pg_hba文件;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}'; '''
        create_msg = self.sh_primy.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication all trust'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ::1/128 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step4:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                        self.constant.GSGUC_SUCCESS_MSG,
                                        f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status("detail")
        self.assertIn("Normal", status, '执行失败:' + text)

    def test_standby_logicreplication(self):
        text = "-----step5:创建表，创建逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
                    create table {self.table_name} \
                    (col_tinyint tinyint ,
                    col_smallint smallint , 
                    col_integer integer, 
                    col_int int, 
                    col_binary_integer binary_integer, 
                    col_bigint bigint, 
                    col_real real, 
                    col_float4 float4, 
                    col_double_precision double precision, 
                    col_float8 float8, 
                    col_float float, 
                    col_float1 float(38), 
                    col_binary_double binary_double, 
                    col_char char, 
                    col_char1 char(50), 
                    col_character character, 
                    col_character1 character(50), 
                    col_varchar varchar, 
                    col_varchar1 varchar(50), 
                    col_character_varying character varying(50), 
                    col_clob clob, 
                    col_text text, 
                    col_blob blob, 
                    col_bytea bytea, 
                    col_date date, 
                    col_time time, 
                    col_time1 time(6), 
                    col_time2 time without time zone, 
                    col_time3 time(6) without time zone, 
                    col_time4 time with time zone, 
                    col_time5 time(6) with time zone, 
                    col_timestamp timestamp, 
                    col_timestamp1 timestamp(6), 
                    col_timestamp2 timestamp without time zone, 
                    col_timestamp3 timestamp(6) without time zone, 
                    col_timestamp4 timestamp with time zone, 
                    col_timestamp5 timestamp(6) with time zone, 
                    col_serial serial, 
                    col_smallserial smallserial, 
                    col_bigserial bigserial);
                    alter table {self.table_name} replica identity default;
                    alter table {self.table_name} \
                    add constraint par_p_k_0001 primary key(col_int);'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertTrue(create_msg.count(self.constant.ALTER_TABLE_MSG) == 3,
                        '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      create_msg, '执行失败:' + text)
        create_cmd = f'''create or replace function {self.function_name}
            (startid int,endid int,tablename varchar) return int is 
            v_tinyint tinyint;
            v_smallint smallint;
            v_integer integer;
            v_int int;
            v_binary_integer binary_integer;
            v_bigint bigint;
            v_real real;
            v_float4 float4;
            v_double_precision double precision;
            v_float8 float8;
            v_float float;
            v_float1 float(38);
            v_binary_double binary_double;
            v_char char;
            v_char_50 char(50);
            v_character character;
            v_character_50 character(50);
            v_varchar varchar;
            v_varchar_50 varchar(50);
            v_character_varying character varying(50);
            v_clob clob;
            v_text text;
            v_blob blob;
            v_bytea bytea;
            v_date date;
            v_time time;
            v_time1 time(6);
            v_time2 time without time zone;
            v_time3 time(6) without time zone;
            v_time4 time with time zone;
            v_time5 time(6) with time zone;
            v_timestamp timestamp;
            v_timestamp1 timestamp(6);
            v_timestamp2 timestamp without time zone;
            v_timestamp3 timestamp(6) without time zone;
            v_timestamp4 timestamp with time zone;
            v_timestamp5 timestamp(6) with time zone;
            begin
            for i in startid..endid
            loop
            v_tinyint :=mod(i,256);
            v_smallint :=i+1;
            v_integer :=i+2;
            v_int :=i+3;
            v_binary_integer :=i+4;
            v_bigint :=i+5;
            v_real :=i*1.11;
            v_float4 :=i*2.22;
            v_double_precision :=i*3.33;
            v_float8 :=i*4.44;
            v_float :=i*5.55;
            v_float1 :=i*6.66;
            v_binary_double :=i*7.77;
            v_char :=substring(''||i,1,1);
            v_char_50 :='wxwhlayyawajbcqzhrctszhddqrwkyzjdwbygz'||i;
            v_character :=substring('abcdefghigklmnopqrstuvwxyz',mod(i,26),1);
            v_character_50 :='v_character_50_length'||i;
            v_varchar :='v_varchar'||i;
            v_varchar_50 :='v_varchar_50'||i;
            v_character_varying :='v_character_varying'||i;
            v_clob :='v_clob'||i;
            v_text :='v_text'||i;
            v_blob :='110101'||mod(i,2)||mod(i+1,2)||mod(i+2,2)||mod(i+3,2)
            ||mod(i+4,2)||mod(i+5,2);
            v_bytea :='100100'||mod(i,2)||mod(i+1,2)||mod(i+2,2)||mod(i,2)
            ||mod(i+1,2)||mod(i+2,2);
            v_date :=(1000+i)||'-01-08';
            v_time :='19:41:'||mod(i,60);
            v_time1 :='20:41:'||mod(i+1,60);
            v_time2 :='21:21:'||mod(i+2,60)||' pst';
            v_time3 :='22:22:'||mod(i+3,60);
            v_time4 :='21:21:'||mod(i+4,60)||' pst';
            v_time5 :='22:22:'||mod(i+5,60);
            v_timestamp :=1000+i||'-4-22';
            v_timestamp1 :=1001+i||'-4-22 pst';
            v_timestamp2 :=1002+i||'-4-22 21:22:23';
            v_timestamp3 :=1003+i||'-4-22 21:22:23.333333';
            v_timestamp4 :=1004+i||'-4-22 pst';
            v_timestamp5 :=1005+i||'-4-22 pst';
            execute immediate 'insert into ' || tablename ||' (col_tinyint,
            col_smallint,col_integer,col_int,col_binary_integer,col_bigint,
            col_real,col_float4,col_double_precision,col_float8,col_float,
            col_float1,col_binary_double,col_char,col_char1,col_character,
            col_character1,col_varchar,col_varchar1,col_character_varying,
            col_clob,col_text,col_blob,col_bytea,col_date,col_time,col_time1,
            col_time2,col_time3,col_time4,col_time5,col_timestamp,
            col_timestamp1,col_timestamp2,col_timestamp3,col_timestamp4,
            col_timestamp5) values
            (:p1,:p2,:p3,:p4,:p5,:p6,:p7,:p8,:p9,:p10,:p11,:p12,:p13,:p14,
            :p15,:p16,:p17,:p18,:p19,:p20,:p21,:p22,:p23,:p24,:p25,:p26,:p27,
            :p28,:p29,:p30,:p31,:p32,:p33,:p34,:p35,:p36,:p37)'
            using v_tinyint,v_smallint,v_integer,v_int,v_binary_integer,
            v_bigint,v_real,v_float4,v_double_precision,v_float8,v_float,
            v_float1,v_binary_double,v_char,v_char_50,v_character,
            v_character_50,v_varchar,v_varchar_50,v_character_varying,v_clob,
            v_text,v_blob,v_bytea,v_date,v_time,v_time1,v_time2,v_time3,
            v_time4,v_time5,v_timestamp,v_timestamp1,v_timestamp2,v_timestamp3,
            v_timestamp4,v_timestamp5;
            end loop;
            return 0;
            end;
            create or replace procedure {self.procedure_name}() is 
                    begin 
                    perform {self.function_name}(1,2,'{self.table_name}');
                    commit;
                    delete from {self.table_name} limit 1;
                    commit;
                    end;'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_FUNCTION_SUCCESS_MSG and
                      self.constant.CREATE_PROCEDURE_SUCCESS_MSG,
                      create_msg, '执行失败:' + text)

        create_slot_1 = f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH};
                          expect <<EOF
                          set timeout 300
                          spawn {create_slot_1}
                          expect "Password:"
                          send "{macro.COMMON_PASSWD}\\n"
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        create_slot_2 = f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-U {self.user_name} " \
            f"-s 2 --create"
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {create_slot_2} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        execute_msg = self.primary_node.sh(execute_cmd).result()
        self.logger.info(execute_msg)
        text = "-----验证创建逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step6:开启逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        primy_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} -s 2 " \
            f"-f {self.log_file_1} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                          expect <<EOF 
                          set timeout 300 
                          spawn {primy_start_cmd} 
                          expect "Password:" 
                          send "{macro.COMMON_PASSWD}\\n" 
                          expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, execute_cmd,))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        msg_result_1 = thread_1.get_result()
        self.logger.info(msg_result_1)
        text = "-----验证主机开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        stanby_start_cmd = f"pg_recvlogical " \
            f"-o skip-empty-xacts=true -F 1 " \
            f"-d postgres -S {self.slot_name_2} " \
            f"-p {self.standby_node.db_port} -s 2 " \
            f"-f {self.log_file_2} " \
            f"-U {self.user_name} " \
            f"--start "
        execute_cmd = f'''source {macro.DB_ENV_PATH}; 
                        expect <<EOF 
                        set timeout 300 
                        spawn {stanby_start_cmd} 
                        expect "Password:" 
                        send "{macro.COMMON_PASSWD}\\n" 
                        expect eof\n''' + '''EOF'''
        self.logger.info(execute_cmd)
        thread_2 = ComThread(self.common.get_sh_result,
                             args=(self.standby_node, execute_cmd,))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_2)
        text = "-----验证备机开启逻辑复制槽成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select slot_name,active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_2 and 't', sql_msg.splitlines()[-2],
                      '执行失败:' + text)

        text = "-----step7:DML操作;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f'''start transaction;
                    call {self.procedure_name}();
                    savepoint s1;
                    delete from {self.table_name} where col_tinyint is not null;
                    rollback to savepoint s1;
                    call {self.function_name}(3, 4, '{self.table_name}');
                    commit;'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertNotIn("ERROR", sql_msg, "执行失败" + text)
        self.assertIn(self.constant.START_TRANSACTION_SUCCESS_MSG and
                      self.procedure_name and "SAVEPOINT" and
                      self.constant.DELETE_SUCCESS_MSG and
                      self.constant.ROLLBACK_MSG and
                      self.function_name and
                      self.constant.COMMIT_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)
        time.sleep(10)

        text = "-----step8:比对数据;expect:成功-----"
        self.logger.info(text)
        shell_cmd_1 = f"sleep 10 && cat {self.log_file_1} "
        self.logger.info(shell_cmd_1)
        shell_msg_1 = self.primary_node.sh(shell_cmd_1).result()
        self.logger.info(shell_msg_1)
        self.assertNotIn("ERROR", shell_msg_1, "执行失败" + text)
        shell_cmd_2 = f"sleep 10 && cat {self.log_file_2} "
        self.logger.info(shell_cmd_2)
        shell_msg_2 = self.standby_node.sh(shell_cmd_2).result()
        self.logger.info(shell_msg_2)
        self.assertNotIn("ERROR", shell_msg_2, "执行失败" + text)
        self.assertEqual(shell_msg_1, shell_msg_2, "执行失败" + text)
        time.sleep(5)

        text = "-----step9:停止逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primy_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primy_result)
        standy_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standy_result)
        time.sleep(5)

    def tearDown(self):
        text = "-----step10:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        drop_slot_cmd_1 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_1} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_1 = self.primary_node.sh(drop_slot_cmd_1).result()
        self.logger.info(drop_slot_msg_1)
        drop_slot_cmd_2 = f"source {macro.DB_ENV_PATH};" \
            f"pg_recvlogical -d postgres " \
            f"-S {self.slot_name_2} " \
            f"-p {self.primary_node.db_port} " \
            f"-s 2 --drop"
        drop_slot_msg_2 = self.primary_node.sh(drop_slot_cmd_2).result()
        self.logger.info(drop_slot_msg_2)
        sql_cmd = f"select slot_name,plugin,slot_type,active," \
            f"restart_lsn from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text = "-----step11:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step11.1:删除表，删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_table_cmd = f'''drop table {self.table_name} cascade;
            drop procedure {self.procedure_name};
            drop function {self.function_name};
            drop user {self.user_name};'''
        drop_table_msg = self.sh_primy.execut_db_sql(drop_table_cmd,
                                                  dbname='postgres')
        self.logger.info(drop_table_msg)

        text_2 = "-----step11.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]];
            then rm -rf {self.postgresql_path};
            mv {self.postgresql_bak_path} {self.postgresql_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        restore_standby_1 = self.standby_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        self.logger.info(restore_standby_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        restore_standby_2 = self.primary_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)
        self.logger.info(restore_standby_2)

        text_3 = "-----step11.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.primary_root.sh(del_cmd)
        self.standby_root.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        standby_file_msg = self.standby_root.sh(file_cmd).result()
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_3)
        self.assertEqual('', drop_slot_msg_1, '执行失败:' + text)
        self.assertEqual('', drop_slot_msg_2, '执行失败:' + text)
        self.assertNotIn(self.slot_name_1 and self.slot_name_2,
                         sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.DROP_FUNCTION_SUCCESS_MSG and
                      self.constant.DROP_PROCEDURE_SUCCESS_MSG and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_table_msg, '执行失败' + text_1)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
